package com.jingwee.springbootkafka.template;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.stereotype.Component;

import java.util.Arrays;
import java.util.concurrent.ExecutionException;

/**
 * @author jingwei
 * @version 1.0
 * @description
 * @date 2023/8/8
 */
@Component
@Slf4j
public class KafkaBaseTemplate {

    private AdminClient adminClient;

    public KafkaBaseTemplate(AdminClient adminClient) {
        this.adminClient = adminClient;
    }

    /**
     * 创建topic
     * @param topicName
     * @return
     */
    public String createTopic(String topicName) {
        NewTopic topic = new NewTopic(topicName, 2, (short) 1);
        CreateTopicsResult createTopicsResult = adminClient.createTopics(Arrays.asList(topic));
        log.info("创建主题结果：isDone={}", createTopicsResult.all().isDone());
        return topicName;
    }

    /**
     * 查询topic
     * @param topicName
     * @return
     */
    public String queryTopic(String topicName) {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topicName));
        StringBuffer sb = new StringBuffer("topic信息:");
        try {
            result.all().get().forEach((k,v)->sb.append("key").append(k).append(";\nv:").append(v));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return sb.toString();
    }

    /**
     * 删除topic
     * @param topicName
     * @return
     */
    public String deleteTopic(String topicName) {
        adminClient.deleteTopics(Arrays.asList(topicName));
        return topicName;
    }
}
